package State

/**
 * 需求:连续两个温度值的温差在十度以内，否则发生报警
 */

import Source.SensorReading
import org.apache.flink.api.common.functions.{RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object StateTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })

    val alertStream = dataStream
      .keyBy(_.id)
      .flatMap(new TempChangeAlert(10.0))

    alertStream.print()

    env.execute("TempChange")
  }
}

/**
 * 自定义Map
 */
class TempChangeAlert(d: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
  //定义状态保存上一次的温度值
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(
    new ValueStateDescriptor[Double]("lastTempStatue", classOf[Double])
  )
  //判断是否十第一个数据
  var flag = true

  override def flatMap(in: SensorReading, collector: Collector[(String, Double, Double)]): Unit = {
    //获取上一个数据的temp值
    val lastTemp = lastTempState.value()
    if (flag) {
      //如果为空，直接更新
      lastTempState.update(in.temperature)
      flag = false
    } else {
      //不为空，判断
      if (Math.abs(lastTemp - in.temperature) >= d) {
        //温差大于十
        collector.collect(("温差大于十", lastTemp, in.temperature))
        //更新
        lastTempState.update(in.temperature)
      } else {
        collector.collect(("温差正常", lastTemp, in.temperature))
        lastTempState.update(in.temperature)
      }
    }
  }
}